package com.zheng.flink.study.dataset.transformation;

import org.apache.flink.api.common.operators.Order;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple3;

import java.util.ArrayList;
import java.util.List;

/**
 * Created by zhengbo on 2019/12/13.
 */
public class DistinctTransformation {

    public static void main(String[] args) throws Exception {

        List<Tuple3<String, Integer, Integer>> personInfoList = new ArrayList<>();

        personInfoList.add(new Tuple3<>("zhangyi", 29, 6000));
        personInfoList.add(new Tuple3<>("zhangyi", 29, 6000));
        personInfoList.add(new Tuple3<>("zhangyi", 30, 6300));
        personInfoList.add(new Tuple3<>("lisi", 25, 6000));
        personInfoList.add(new Tuple3<>("wangsu", 27, 6000));
        personInfoList.add(new Tuple3<>("zxxy", 29, 20000));

        ExecutionEnvironment env = ExecutionEnvironment.getExecutionEnvironment();

        DataSource<Tuple3<String, Integer, Integer>> dataSource = env.fromCollection(personInfoList);
        //对tuple中字段完全一致的去重
        dataSource.distinct().print();

        //先按年龄排序 再对工资相同的去重
        dataSource.sortPartition(1, Order.ASCENDING).distinct(2).print();

    }
}
